package model

import (
	"fmt"
	"qiniupkg.com/x/errors.v7"
)

var (
	allProcessor map[string]*Processor
	allPools     map[string]*Pool
	allChannels  map[string]*Channel
	allProChan   chan struct{}
)

func init() {
	allPools = make(map[string]*Pool)
	allProChan = make(chan struct{})
	allChannels = make(map[string]*Channel)
	allProcessor = make(map[string]*Processor)
}

var (
	errPoolNotExist      = errors.New("pool not exist")
	errChannelNotExist   = errors.New("channel not exist")
	errProcessorNotExist = errors.New("processor not exist")
	errProcessorHasStop  = errors.New("processor has stop")
)

func AddProcessor(appName string, parallelMaxMsg int) {
	//add processor
	allProcessor[appName] = NewProcessor(appName, parallelMaxMsg)
	//add Pool
	allPools[appName] = NewTaskPool(appName)
	//add Channel
	allChannels[appName] = NewChannel(appName)
}

func StartAllProcessor() {
	for name, processor := range allProcessor {
		go processor.Do()
		//monitor
		ch, _ := GetChannel(name)
		pool, _ := GetPool(name)
		go pool.Monitor(ch)
		//warning task status
		go pool.Supervisor(name)
	}
}

func StopAllProcessor() {
	for name, p := range allProcessor {
		fmt.Printf("[execute]: processor [appName:%v] stop\n", name)
		p.Stop()
		fmt.Printf("[end]: processor [appName:%v] stop success\n", name)
	}
}

func GetProcessor(appName string) (*Processor, error) {
	if p, ok := allProcessor[appName]; ok && p.Valid() {
		return p, nil
	}
	return nil, errProcessorNotExist
}

func GetPool(appName string) (*Pool, error) {
	if p, ok := allPools[appName]; ok {
		return p, nil
	}
	return nil, errPoolNotExist
}

func GetChannel(appName string) (*Channel, error) {
	if c, ok := allChannels[appName]; ok {
		return c, nil
	}
	return nil, errChannelNotExist
}

func GetAllPool() map[string]*Pool {
	return allPools
}

func RecoverTask() {
	//pool recover
	for appName, pool := range GetAllPool() {
		ch, err := GetChannel(appName)
		if err != nil {
			panic("server dispatcher channel error")
		}
		//recover
		pool.Recover(appName, ch)
	}
}

//total processor stop working
func CloseProChan() {
	close(allProChan)
}

func GetProChan() <-chan struct{} {
	return allProChan
}
